package net.library.clink.impl;

import net.library.clink.core.IoProvider;
import net.library.clink.utils.CloseUtils;

import java.io.IOException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

public class IoSelectorProvider implements IoProvider {
    // 标识是否已经被关闭了
    private final AtomicBoolean isClosed = new AtomicBoolean(false);

    // 锁：判断是否处于某个过程中(注册输入或者注册输出）
    private final AtomicBoolean inRegInput = new AtomicBoolean(false);
    private final AtomicBoolean inRegOutput = new AtomicBoolean(false);

    // 把读和写的selector分开
    private final Selector readSelector;
    private final Selector writeSelector;

    // 把SelectionKey和对应的通知函数关联起来
    private final HashMap<SelectionKey, Runnable> inputCallbackMap = new HashMap<>();
    private final HashMap<SelectionKey, Runnable> outputCallbackMap = new HashMap<>();

    // 读和写的处理线程池也分开
    private final ExecutorService inputHandlePool;
    private final ExecutorService outputHandlePool;

    public IoSelectorProvider() throws IOException {
        this.readSelector = Selector.open();
        this.writeSelector = Selector.open();

        inputHandlePool = Executors.newFixedThreadPool(4, new IoProviderThreadFactory("IoProvider-Input-Thread-"));
        outputHandlePool = Executors.newFixedThreadPool(4, new IoProviderThreadFactory("IoProvider-Output-Thread-"));

        // 开始输入和输出的(启动)监听
        startRead();
        startWrite();
    }

    private void startRead() {
        Thread thread = new Thread("Clink IoSelectorProvider ReadSelector Thread") {
            @Override
            public void run() {
                while (!isClosed.get()) {
                    try {
                        if (readSelector.select() == 0) {
                            waitSelection(inRegInput);
                            continue;
                        }
                        Set<SelectionKey> selectionKeys = readSelector.selectedKeys();
                        //======================== Iterator方法==========================================
                        Iterator<SelectionKey> iterator = selectionKeys.iterator();
                        while (iterator.hasNext()) {
                            SelectionKey selectionKey = iterator.next();
                            if (selectionKey.isValid()) {
                                handleSelection(selectionKey, SelectionKey.OP_READ, inputCallbackMap, inputHandlePool);
                            }
                            // 在遍历集合的过程中，如何调用remove方法，会报ConcurrentModificationException
                            //selectionKeys.remove(selectionKey);
                        }
                        //======================== For-Each方法 =========================================
//                        for (SelectionKey selectionKey : selectionKeys) {
//                            // 处理有效的SelectionKey， 一个SelectionKey是一个Channel和一个Selector注册关系的表示token
//                            // 因为键有可能被取消，而被移入到cancelledKeys中，所以需要先用isValid判断
//                            if (selectionKey.isValid()) {
//                                // 在handleSelection里第一步就是进行反注册，即在异步处理某个selectionKey时不需要在后面的循环中继续判断该SelectionKey
//                                // 等到handle处理完毕，在重新注册OP_READ事件
//                                handleSelection(selectionKey, SelectionKey.OP_READ, inputCallbackMap, inputHandlePool);
//                            }
//                            // 在遍历集合的过程中，如何调用remove方法，会报ConcurrentModificationException
//                            //selectionKeys.remove(selectionKey);
//                        }
                        //System.out.println("有数据需要读取：" + selectionKeys.size());
                        //============================ 关键操作 ==========================================
                        // 这一句有点关键，之前少了这一句，客户端只能发送的第一条数据被服务器端接收到
                        // 为什么需要清除这个集合呢?
                        // 如果不清除该集合，则上面的readSelector.select()操作，永远都会返回0，而不是处于阻塞状态，等待通道对应的IO事件的到来
                        //********************************
                        selectionKeys.clear();
                        //********************************
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        thread.setPriority(Thread.MAX_PRIORITY);
        thread.start();
    }

    private void startWrite() {
        Thread thread = new Thread("Clink IoSelectorProvider WriteSelector Thread") {
            @Override
            public void run() {
                while (!isClosed.get()) {
                    try {
                        if (writeSelector.select() == 0) {
                            waitSelection(inRegOutput);
                            continue;
                        }
                        Set<SelectionKey> selectionKeys = writeSelector.selectedKeys();
                        for (SelectionKey selectionKey : selectionKeys) {
                            // 处理有效的SelectionKey， 一个SelectionKey是一个Channel和一个Selector注册关系的表示token
                            if (selectionKey.isValid()) {
                                // 在handleSelection里第一步就是进行反注册，即在异步处理某个selectionKey时不需要在后面的循环中继续判断该SelectionKey
                                // 等到handle处理完毕，再重新注册OP_WRITE事件
                                handleSelection(selectionKey, SelectionKey.OP_WRITE, outputCallbackMap, outputHandlePool);
                            }
                        }
                        selectionKeys.clear();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        thread.setPriority(Thread.MAX_PRIORITY);
        thread.start();
    }

    static class IoProviderThreadFactory implements ThreadFactory {
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        IoProviderThreadFactory(String namePrefix) {
            SecurityManager s = System.getSecurityManager();
            this.group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
            this.namePrefix = namePrefix;
        }


        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
            if (t.isDaemon()) {
                t.setDaemon(false);
            }
            if (t.getPriority() != Thread.NORM_PRIORITY) {
                t.setPriority(Thread.NORM_PRIORITY);
            }
            return t;
        }
    }

    @Override
    public boolean registerInput(SocketChannel channel, HandleInputCallback callback) {
        // 将channel注册到Selector
        return registerSelection(channel, readSelector, SelectionKey.OP_READ, inRegInput, inputCallbackMap, callback) != null;
    }

    @Override
    public boolean registerOutput(SocketChannel channel, HandleOutputCallback callback) {
        return registerSelection(channel, writeSelector, SelectionKey.OP_WRITE, inRegOutput, outputCallbackMap, callback) != null;
    }

    @Override
    public void unRegisterInput(SocketChannel channel) {
        unRegisterSelection(channel, readSelector, inputCallbackMap);
    }


    @Override
    public void unRegisterOutput(SocketChannel channel) {
        unRegisterSelection(channel, writeSelector, outputCallbackMap);
    }


    @Override
    public void close() throws IOException {
        if (isClosed.compareAndSet(false, true)) {
            inputHandlePool.shutdown();
            outputHandlePool.shutdown();

            inputCallbackMap.clear();
            outputCallbackMap.clear();

            readSelector.wakeup();
            writeSelector.wakeup();

            CloseUtils.close(readSelector, writeSelector);
        }
    }

    private static void waitSelection(final AtomicBoolean locker) {
        synchronized (locker) {
            if (locker.get()) {
                try {
                    locker.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private static void unRegisterSelection(SocketChannel channel, Selector selector, Map<SelectionKey, Runnable> map) {
        if (channel.isRegistered()) {
            SelectionKey key = channel.keyFor(selector);
            if (key != null) {
                // 取消监听的方法
                key.cancel();
                map.remove(key);
                selector.wakeup();
            }
        }
    }

    private static SelectionKey registerSelection(SocketChannel channel, Selector selector, int registerOps, AtomicBoolean locker,
                          HashMap<SelectionKey, Runnable> map, Runnable runnable) {

        synchronized (locker) {
            // 设置状态：当前处于某个注册过程中
            locker.set(true);

            try {
                // 唤醒当前的selector， 让selector不处于select()的阻塞状态
               selector.wakeup();
               SelectionKey key = null;
               if (channel.isRegistered()) {
                   // 查询是否已经注册过
                   key = channel.keyFor(selector);
                   if (key != null) {
                       // 若之前已注册，则再注册一个新事件registerOps
                       key.interestOps(key.readyOps() | registerOps);
                   }
               }

               if (key == null) {
                   // 注册selector得到SelectionKey
                   key = channel.register(selector, registerOps);
                   // 注册和该SelecitonKey绑定的回调函数
                   map.put(key, runnable);
               }

               return key;
            } catch (ClosedChannelException e) {
                return null;
            } finally {
                // 解除锁定状态
                locker.set(false);
                try {
                    // 通知
                    locker.notify();
                } catch (Exception ignored) {

                }
            }
        }
    }

    private static void handleSelection(SelectionKey key, int keyOps, HashMap<SelectionKey, Runnable> map, ExecutorService pool) {
        // 重点：反注册
        // 取消继续对KeyOps的监听，在这里取消了监听后，就不会存在消息到达提醒重复触发的情况。
        key.interestOps(key.readyOps() & ~keyOps);
        Runnable runnable = null;
        try {
            runnable = map.get(key);

        } catch (Exception ignored) {
        }

        if (runnable != null && !pool.isShutdown()) {
            // 异步调度
            pool.execute(runnable);
        }
    }
}
